/*! ******************************************************************************
*
* Pentaho Data Integration
*
* Copyright (C) 2002-2017 by Hitachi Vantara : http://www.pentaho.com
*
*******************************************************************************
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
******************************************************************************/

package org.pentaho.di.ui.trans.steps.shentong.oss;

import java.util.UUID;

import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowDataUtil;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.util.Utils;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.di.ui.trans.steps.shentong.oss.client.ObjectUtils;
import org.pentaho.di.ui.trans.steps.shentong.oss.client.ResponseBody;

/**
 * This class is part of the demo step plug-in implementation.
 * It demonstrates the basics of developing a plug-in step for PDI. 
 * 
 * The demo step adds a new string field to the row stream and sets its
 * value to "Hello World!". The user may select the name of the new field.
 *   
 * This class is the implementation of StepInterface.
 * Classes implementing this interface need to:
 * 
 * - initialize the step
 * - execute the row processing logic
 * - dispose of the step 
 * 
 * Please do not create any local fields in a StepInterface class. Store any
 * information related to the processing logic in the supplied step data interface
 * instead.  
 * 
 */

public class ShenTongOssStep extends BaseStep implements StepInterface {

	private static final Class<?> PKG = ShenTongOssStepMeta.class; // for i18n purposes
	private ShenTongOssStepMeta meta;
	private ShenTongOssStepData data;

	/**
	 * The constructor should simply pass on its arguments to the parent class.
	 * 
	 * @param s                 step description
	 * @param stepDataInterface step data class
	 * @param c                 step copy
	 * @param t                 transformation description
	 * @param dis               transformation executing
	 */
	public ShenTongOssStep(StepMeta s, StepDataInterface stepDataInterface, int c, TransMeta t, Trans dis) {
		super(s, stepDataInterface, c, t, dis);
	}

	/**
	 * This method is called by PDI during transformation startup.
	 * 
	 * It should initialize required for step execution.
	 * 
	 * The meta and data implementations passed in can safely be cast to the step's
	 * respective implementations.
	 * 
	 * It is mandatory that super.init() is called to ensure correct behavior.
	 * 
	 * Typical tasks executed here are establishing the connection to a database, as
	 * wall as obtaining resources, like file handles.
	 * 
	 * @param smi step meta interface implementation, containing the step settings
	 * @param sdi step data interface implementation, used to store runtime
	 *            information
	 * 
	 * @return true if initialization completed successfully, false if there was an
	 *         error preventing the step from working.
	 * 
	 */
	public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
		// Casting to step-specific implementation classes is safe
		ShenTongOssStepMeta meta = (ShenTongOssStepMeta) smi;
		ShenTongOssStepData data = (ShenTongOssStepData) sdi;
		if (!super.init(meta, data)) {
			return false;
		}

		// Add any step-specific initialization that may be needed here
		return true;
	}

	/**
	 * Once the transformation starts executing, the processRow() method is called
	 * repeatedly by PDI for as long as it returns true. To indicate that a step has
	 * finished processing rows this method must call setOutputDone() and return
	 * false;
	 * 
	 * Steps which process incoming rows typically call getRow() to read a single
	 * row from the input stream, change or add row content, call putRow() to pass
	 * the changed row on and return true. If getRow() returns null, no more rows
	 * are expected to come in, and the processRow() implementation calls
	 * setOutputDone() and returns false to indicate that it is done too.
	 * 
	 * Steps which generate rows typically construct a new row Object[] using a call
	 * to RowDataUtil.allocateRowData(numberOfFields), add row content, and call
	 * putRow() to pass the new row on. Above process may happen in a loop to
	 * generate multiple rows, at the end of which processRow() would call
	 * setOutputDone() and return false;
	 * 
	 * @param smi the step meta interface containing the step settings
	 * @param sdi the step data interface that should be used to store
	 * 
	 * @return true to indicate that the function should be called again, false if
	 *         the step is done
	 */
  	public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException {
		meta = (ShenTongOssStepMeta) smi;
		data = (ShenTongOssStepData) sdi;

		Object[] r = getRow(); // get row from rowset, wait for our turn, indicate busy!
		if (r == null) { // no more input to be expected...

			setOutputDone();
			return false;
		}

		boolean ok = write2Oss(getInputRowMeta(), r);
		if (!ok) {
			setOutputDone();
			return false;
		}

		if (checkFeedback(getLinesRead())) {
			if (log.isDetailed()) {
				if (log.isDetailed()) {
					logBasic(BaseMessages.getString(PKG, "SplitFieldToRows.Log.LineNumber") + getLinesRead());
				}
			}
		}
		return true;
	}
  
	private boolean write2Oss(RowMetaInterface rowMeta, Object[] rowData) throws KettleException {
		if (first) {
			first = false;

			data.outputRowMeta = getInputRowMeta().clone();
			meta.getFields(data.outputRowMeta, getStepname(), null, null, this, repository, metaStore);

			String realBinaryFieldName = environmentSubstitute(meta.getBinaryField());
			data.fieldnr = rowMeta.indexOfValue(realBinaryFieldName);
			
			String realFileNameFieldName = environmentSubstitute(meta.getFileNameField());
			data.fieldnr1 = rowMeta.indexOfValue(realFileNameFieldName);

			int numErrors = 0;
			if (Utils.isEmpty(meta.getDownloadField())) {
				logError(BaseMessages.getString(PKG, "ShenTongOss.Log.DownloadFieldNameIsNull"));
				numErrors++;
			}

			if (data.fieldnr < 0) {
				logError(BaseMessages.getString(PKG, "ShenTongOss.Log.CouldNotFindBinaryField", realBinaryFieldName));
				numErrors++;
			}
			
			if (!rowMeta.getValueMeta(data.fieldnr).isBinary()) {
				logError(BaseMessages.getString(PKG, "ShenTongOss.Log.BinaryFieldNotValid", realBinaryFieldName));
				numErrors++;
			}

			if (numErrors > 0) {
				setErrors(numErrors);
				stopAll();
				return false;
			}

			data.binaryMeta = rowMeta.getValueMeta(data.fieldnr);
			data.fileNameMeta = rowMeta.getValueMeta(data.fieldnr1);
		}
		
		byte[] originalBinary = data.binaryMeta.getBinary(rowData[data.fieldnr]);
		if (originalBinary == null) {
			originalBinary = new byte[0];
		}
		
		String originalFileName;
		if (data.fieldnr1 < 0) {
			originalFileName = UUID.randomUUID().toString();
		} else {
			originalFileName = data.fileNameMeta.getString(rowData[data.fieldnr1]);
			if (originalFileName == null || "".equals(originalFileName)) {
				originalFileName = UUID.randomUUID().toString();
			} else {
				originalFileName = UUID.randomUUID().toString() + 
						(-1 != originalFileName.indexOf(".") ? originalFileName.substring(originalFileName.lastIndexOf(".")) : "");
			}
		}
		String host = environmentSubstitute(meta.getUploadHost());
		String user = environmentSubstitute(meta.getUser());
		String pwd = environmentSubstitute(meta.getPassword());
		String bucket = environmentSubstitute(meta.getBucket());
		
		String downloadPath = null;
		@SuppressWarnings("rawtypes")
		ResponseBody responseBody = ObjectUtils.putObject(host, user, pwd, bucket, originalFileName, originalBinary);
        int statusCode = responseBody.getStatusCode();
        if (statusCode != 200 && statusCode != 204) {
            logBasic("连接OSS配置：【{0},{1},{2},{3}】", host, user, pwd, bucket);
            logError("上传文件【{0}】至OSS错误：{1}", originalFileName, responseBody.getMessage());
            throw new KettleException("上传文件至OSS错误：" + responseBody.getMessage());
        }
    	downloadPath = ObjectUtils.generateDownloadURL(host, user, pwd, bucket, originalFileName);
        if (downloadPath == null) {
        	logError("获取文件下载地址错误：【{0},{1},{2},{3},{4}】", host, user, pwd, bucket, originalFileName);
        	throw new KettleException("获取文件下载地址错误：" + responseBody.getMessage());
        }
        
		Object[] outputRow = RowDataUtil.createResizedCopy(rowData, data.outputRowMeta.size());
		outputRow[rowMeta.size()] = downloadPath;
		putRow(data.outputRowMeta, outputRow);
		return true;
	}

	/**
	 * This method is called by PDI once the step is done processing.
	 * 
	 * The dispose() method is the counterpart to init() and should release any
	 * resources acquired for step execution like file handles or database
	 * connections.
	 * 
	 * The meta and data implementations passed in can safely be cast to the step's
	 * respective implementations.
	 * 
	 * It is mandatory that super.dispose() is called to ensure correct behavior.
	 * 
	 * @param smi step meta interface implementation, containing the step settings
	 * @param sdi step data interface implementation, used to store runtime
	 *            information
	 */
	public void dispose(StepMetaInterface smi, StepDataInterface sdi) {

		// Casting to step-specific implementation classes is safe
		ShenTongOssStepMeta meta = (ShenTongOssStepMeta) smi;
		ShenTongOssStepData data = (ShenTongOssStepData) sdi;

		// Add any step-specific initialization that may be needed here

		// Call superclass dispose()
		super.dispose(meta, data);
	}
}
